package com.hudsonmq.spring.impl.produce.helper;

import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.Duration;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.concurrent.CallBack;
import com.hudson.core.json.JSONUtil;
import com.hudsonmq.spring.HudsonMQClientController;
import com.hudsonmq.spring.domain.produce.TopicQueueInfo;
import com.hudsonmq.spring.utils.NettyUtil;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/27 11:41
 **/
public class SendMessageHelper {
    private static final ConcurrentHashMap<String, AtomicInteger> TOPIC_INDEX_MAP = new ConcurrentHashMap<>();
    private static final Logger log = LoggerFactory.getLogger(SendMessageHelper.class);
    private static ScheduledThreadPoolExecutor retry_service = new ScheduledThreadPoolExecutor(4, r -> new Thread(r, ""));
    private static ConcurrentHashMap<String, Long> WAIT_TIME_MAP = new ConcurrentHashMap<>();

    static {
        retry_service.scheduleAtFixedRate(() -> {
            List<String> topicList = new ArrayList<>();
            long now = System.currentTimeMillis();
            for (Map.Entry<String, Long> entry : WAIT_TIME_MAP.entrySet()) {
                if (now - entry.getValue() > 5000) {
                    topicList.add(entry.getKey());
                }
            }
            Map<String, Channel> channelMap = HudsonMQClientController.BROKER_CHANNEL_MAP;
            for (Map.Entry<String, Channel> channelEntry : channelMap.entrySet()) {
                channelEntry.getValue().writeAndFlush(RemotingCommand.builder()
                        .type(MessageTypeEnum.ADD_TOPIC_QUEUE.getType())
                        .body(JSONUtil.toJson(topicList).getBytes(StandardCharsets.UTF_8))
                        .build()).addListener(f -> {
                    if (f.isSuccess()) {
                        log.debug("{} 添加topic请求成功", channelEntry.getKey());
                    } else {
                        log.error("{} 添加topic请求失败", channelEntry.getKey());
                    }
                });
            }
        }, 0, 5, TimeUnit.SECONDS);
    }



    public static TopicQueueInfo sendMessage(long messageId, Message message, CallBack callBack) {

        TopicQueueInfo topicQueue = TopicQueueHelper.getTopicQueue(message.getTopic(),
                TOPIC_INDEX_MAP.computeIfAbsent(message.getTopic(), k -> new AtomicInteger(0)).getAndIncrement());
        if (topicQueue == null) {
            retry_service.schedule(() -> {
                try {
                    if (!WAIT_TIME_MAP.containsKey(message.getTopic())) {
                        WAIT_TIME_MAP.put(message.getTopic(), System.currentTimeMillis());
                    }
                    sendMessage(messageId, message, callBack);
                } catch (Exception e) {
                    log.error("error");
                }
            }, 2, TimeUnit.SECONDS);
        } else {
            Channel chanel = HudsonMQClientController.BROKER_CHANNEL_MAP.get(topicQueue.getBrokerName());
            message.setQueueId(topicQueue.getQueueId());
            NettyUtil.write(chanel, RemotingCommand.builder()
                    .type(MessageTypeEnum.SEND_MESSAGE_REQ.getType())
                    .opaque(messageId)
                    .body(com.hudson.codec.domain.Message.serialize(message))
                    .build()).addListener(f -> {
                if (f.isSuccess()) {
                    log.debug("消息 {} 發送成功", messageId);
                    if (callBack != null) {
                        SendMessageSyncHelper.put(messageId, callBack);
                    }
                } else {
                    log.error("消息{}:{}发送失败", messageId, new String(message.getBody(), StandardCharsets.UTF_8));
                }
            });
        }
        return topicQueue;
    }

    public static TopicQueueInfo sendRelayMessage(long messageId, Message message, Duration duration) {

        TopicQueueInfo topicQueue = TopicQueueHelper.getTopicQueue(message.getTopic(),
                TOPIC_INDEX_MAP.computeIfAbsent(message.getTopic(), k -> new AtomicInteger(0)).getAndIncrement());
        if (topicQueue == null) {
            retry_service.schedule(() -> {
                try {
                    if (!WAIT_TIME_MAP.containsKey(message.getTopic())) {
                        WAIT_TIME_MAP.put(message.getTopic(), System.currentTimeMillis());
                    }
                    sendRelayMessage(messageId, message, duration);
                } catch (Exception e) {
                    log.error("error");
                }
            }, 2, TimeUnit.SECONDS);
        } else {
            Channel chanel = HudsonMQClientController.BROKER_CHANNEL_MAP.get(topicQueue.getBrokerName());

            Map<String, String> map = new HashMap<>();
            map.put(Duration.class.getSimpleName(), JSONUtil.toJson(duration));


            message.setQueueId(topicQueue.getQueueId());
            message.setProperties(map);

            NettyUtil.write(chanel, RemotingCommand.builder()
                    .type(MessageTypeEnum.SEND_RELAY_MESSAGE_REQ.getType())
                    .opaque(messageId)
                    .body(Message.serialize(message))
                    .build());
        }
        return topicQueue;
    }


}
